package io.kiki.sba.registry.server.data_store.multi.cluster.storage;


import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.common.model.metaserver.RemoteDatumClearEvent;
import io.kiki.sba.registry.common.model.metaserver.RemoteDatumClearEvent.DatumType;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.change.DataChangeType;
import io.kiki.sba.registry.server.data_store.slot.SlotAccessorDelegate;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.Map;


public class MultiClusterDatumService {

    private static final Logger logger = LoggerFactory.getLogger(MultiClusterDatumService.class);

    @Autowired
    private DatumStorageDelegate datumStorageDelegate;

    @Autowired
    private SlotAccessorDelegate slotAccessorDelegate;

    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;

    @Autowired
    private DataChangeEventCenter dataChangeEventCenter;

    public void clear(RemoteDatumClearEvent request) {

        MultiClusterSyncInfo query = multiClusterSyncRepository.query(request.getRemoteDataCenter());
        if (query != null && query.isEnableSyncDatum()) {
            logger.error("clear datum forbidden when sync enable, request:{}", request);
            return;
        }

        Map<String, DatumVersion> datumVersionMap = new HashMap<>();
        if (request.getDatumType() == DatumType.data_info_id) {
            if (!slotAccessorDelegate.isLeader(request.getRemoteDataCenter(), slotAccessorDelegate.slotOf(request.getDataInfoId()))) {
                return;
            }
            DatumVersion datumVersion = datumStorageDelegate.clearPublishers(request.getRemoteDataCenter(), request.getDataInfoId());
            if (datumVersion != null) {
                datumVersionMap.put(request.getDataInfoId(), datumVersion);
            }
        } else
            if (request.getDatumType() == DatumType.group) {
                datumVersionMap = datumStorageDelegate.clearGroupPublishers(request.getRemoteDataCenter(), request.getGroup());
            } else {
                throw new SofaRegistryRuntimeException("illegal request datumType:" + request.getDatumType());
            }

        logger.info("[ClearMultiDatum]req: {}, datumVersionMap:{}", request, datumVersionMap);
        if (!CollectionUtils.isEmpty(datumVersionMap)) {
            dataChangeEventCenter.onChange(datumVersionMap.keySet(), DataChangeType.clear, request.getRemoteDataCenter());
        }
    }
}
